package com.cn.file.ha;

import com.cn.coder.MessageDecoder;
import com.cn.coder.MessageEncoder;
import com.cn.common.BrokerEntity;
import com.cn.common.RemoteHandler;
import com.cn.common.ServerThread;
import com.cn.common.WaitNotifyObject;
import com.cn.common.processor.INettyRequestProcessor;
import com.cn.file.CommitLog;
import com.cn.file.DefaultMessageStore;
import com.cn.msg.Message;
import com.cn.remote.ChannelListener;
import com.cn.remote.RemoteServer;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.handler.timeout.IdleStateHandler;

import java.nio.charset.Charset;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;

public class HAService {

    private CommitLog commitLog;

    private WaitNotifyObject waitNotifyObject = new WaitNotifyObject();

    private DefaultMessageStore defaultMessageStore;

    private final RemoteServer remoteServer;

    private TransferService transferService;

    private List<HAConnection> haConnections = new ArrayList<>();
    private Map<Channel, HAConnection> haMaps = new HashMap<>();
    private ChannelListener channelListener;
    private INettyRequestProcessor nettyRequestProcessor;
    private boolean slaveOk = false;
    private HAClient haClient;
    private long offset = -1;

    private ExecutorService executorService;
    private ScheduledExecutorService timerTask = Executors.newScheduledThreadPool(1);
    private BlockingQueue<Runnable> HABlock;

    public HAService(DefaultMessageStore defaultMessageStore, CommitLog commitLog) {
        this.HABlock = new LinkedBlockingQueue<>(100);
        this.executorService = new ThreadPoolExecutor(4, 16, 1 * 60, TimeUnit.SECONDS, HABlock);

        this.defaultMessageStore = defaultMessageStore;
        this.commitLog = commitLog;


        this.transferService = new TransferService();
        int port = (this.defaultMessageStore.getBrokerController().getPort() - 1);
        this.remoteServer = new RemoteServer(1, "0.0.0.0:" + port);
        nettyRequestProcessor = new HACopyData(this);
        channelListener = new ChannelRegisterImpl();
        this.remoteServer.registerChannelListener(channelListener);
        this.remoteServer.registerProcessor(20, nettyRequestProcessor, executorService);

        this.haClient = new HAClient();
        start();
    }

    public void start() {
        //启动服务端
        this.remoteServer.start();

        //id==0 是master，它不需要启动client
        if (this.defaultMessageStore.getBrokerController().getBroker_id() != 0) {
            //启动客户端
            timerTask.scheduleAtFixedRate(new Runnable() {
                @Override
                public void run() {
                    try {
                        haClient.connection();
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            }, 10, 10, TimeUnit.SECONDS);

            haClient.start();
        }
        transferService.start();

    }

    public WaitNotifyObject getWaitNotifyObject() {
        return waitNotifyObject;
    }

    public void shutdown() {
        this.remoteServer.shutdown();
    }

    public void addRequest(CommitLog.GroupCommitRequest request) {
        this.transferService.addRequest(request);
    }

    public boolean isSlaveOk() {
        return slaveOk;
    }

    public void setSlaveOk(boolean slaveOk) {
        this.slaveOk = slaveOk;
    }

    public long getOffset() {
        return offset;
    }

    public void setOffset(long offset) {
        this.offset = offset;
    }

    public CommitLog getCommitLog() {
        return commitLog;
    }

    public DefaultMessageStore getDefaultMessageStore() {
        return defaultMessageStore;
    }

    public long getCommitOffset() {
        if (this.commitLog != null) {
            return this.commitLog.getMaxOffset();
        } else {
            return -1;
        }
    }

    //向slave节点拷贝数据
    class TransferService extends ServerThread {

        private WaitNotifyObject waitNotifyObject = new WaitNotifyObject();
        private volatile List<CommitLog.GroupCommitRequest> writeList = new ArrayList<>();
        private volatile List<CommitLog.GroupCommitRequest> readList = new ArrayList<>();

        public synchronized void addRequest(CommitLog.GroupCommitRequest request) {
            synchronized (this.writeList) {
                this.writeList.add(request);
            }
            if (hasNotified.compareAndSet(false, true)) {
                //通知有新的请求到了
                waitPoint.countDown();
            }
        }

        @Override
        public String getServiceName() {
            return TransferService.class.getName();
        }

        @Override
        public void run() {
            System.out.println("向salve节点复制数据");
            while (!isStopped()) {
                waitForRunning(10);
                try {
                    transfer();
                } catch (Exception e) {
                    //处理异常
                }
            }
        }

        public void transfer() {
            if (this.readList != null && this.readList.size() > 0) {
                synchronized (this.readList) {
                    for (CommitLog.GroupCommitRequest request : this.readList) {
                        boolean flushOk = false;
                        this.waitNotifyObject.waitRunning(200);
                        request.wakeupCustomer(flushOk);
                    }
                    this.readList.clear();
                }
            }
        }

        public void swapRequest() {
            List<CommitLog.GroupCommitRequest> tmp = this.writeList;
            this.writeList = this.readList;
            this.readList = tmp;
        }

        @Override
        protected void onWaitEnd() {
            swapRequest();
        }
    }

    class HAClient extends ServerThread {
        private EventLoopGroup worker;
        private Bootstrap bootstrap;
        private String address;
        private Channel channel;


        public HAClient() {
            this.worker = new NioEventLoopGroup();
            this.bootstrap = new Bootstrap();
            this.bootstrap.group(worker);
            this.bootstrap.channel(NioSocketChannel.class);
            this.bootstrap.option(ChannelOption.TCP_NODELAY, true);
            this.bootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 3000);
            this.bootstrap.handler(new HAClientHandler());
        }

        public void start() {
            super.start();
        }

        public void connection() {
            if (address == null) {
                BrokerEntity brokerEntity = defaultMessageStore.getMaster();
                if (brokerEntity != null && brokerEntity.getBroker_id() == 0) {
                    String as[] = brokerEntity.getAddress().split(":");
                    address = as[0];
                    int port = Integer.valueOf(as[1]);
                    port = port - 1;
                    address = address + ":" + port;
                }
            }
            if (!isSlaveOk()) {
                if (address != null) {
                    ChannelFuture f = bootstrap.connect(RemoteHandler.string2SocketAddress(address));
                    if (f.awaitUninterruptibly(1000, TimeUnit.MILLISECONDS)) {
                        System.out.println("连接master成功");
                        //连接成功
                        setSlaveOk(true);
                        channel = f.channel();
                        //上报位移
                        reportOffset();
                    } else {
                        setSlaveOk(false);
                    }
                }

            }
        }

        //上报位移
        public void reportOffset() {
            try {
                offset = getCommitOffset();
                if (offset == -1) {
                    return;
                }
                Message message = new Message();
                message.setType((byte) 20);//上报位移
                message.setFlag((byte) 2);
                message.setNettyRequestWrapperCode((byte) 20);
                byte[] bytes = String.valueOf(offset).getBytes(Charset.forName("utf-8"));
                message.setByteBodys(bytes);
                message.setLength(message.getByteBodys().length);
                if (this.channel != null && this.channel.isActive()) {
                    this.channel.writeAndFlush(message);
                }
            } catch (Exception e) {
                e.printStackTrace();
                System.out.println("上报位移错误");
            }
        }

        @Override
        public String getServiceName() {
            return HAClient.class.getName();
        }

        @Override
        public void run() {
            //上报位移
            while (!isStopped()) {
                reportOffset();//上报
                waitForRunning(30000);//等待30秒
            }
        }
    }

    class ChannelRegisterImpl implements ChannelListener {
        @Override
        public void registerEvent(ChannelHandlerContext ctx) {
            HAConnection haConnection = new HAConnection(HAService.this, ctx);
            haConnections.add(haConnection);
            haMaps.put(ctx.channel(), haConnection);
            System.out.println("注册了：" + haMaps);
        }

        @Override
        public void unregisterEvent(ChannelHandlerContext ctx) {
            HAConnection haConnection = haMaps.remove(ctx.channel());
            if (null != haConnection) {
                haConnections.remove(haConnection);
                haConnection.shutdown();
                haConnection = null;
            }
            System.out.println("取消订阅：" + haMaps);
        }
    }


    class HAClientHandler extends ChannelInitializer<SocketChannel> {

        @Override
        protected void initChannel(SocketChannel ch) throws Exception {
            //如果消息是自定义类型，那么用以下处理器
            ch.pipeline().addLast(new MessageEncoder());//编码器
            ch.pipeline().addLast(new MessageDecoder());//解码器
            ch.pipeline().addLast(new IdleStateHandler(0, 0, 3, TimeUnit.SECONDS));//心跳设置
            ch.pipeline().addLast(new HAClientBeat());//心跳处理器
            ch.pipeline().addLast(new HAClientProcessor());//业务处理器
        }
    }

    class HAClientBeat extends ChannelDuplexHandler {

        @Override
        public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
            if (evt instanceof IdleStateEvent) {
                IdleState state = ((IdleStateEvent) evt).state();
                if (state.equals(IdleState.ALL_IDLE)) {
                    Message message = new Message();
                    message.setType((byte) 1);//这里需要区分是 心跳包类型
                    message.setFlag((byte) 2);
                    message.setByteBodys("心跳".getBytes("UTF-8"));
                    ctx.writeAndFlush(message);
                }
            } else {
                super.userEventTriggered(ctx, evt);
            }
        }
    }

    /**
     * 自定义消息处理器
     */
    class HAClientProcessor extends ChannelInboundHandlerAdapter {

        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {

        }

        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) {
            System.out.println("收到新的消息了");
            if (msg != null) {
                Message message = (Message) msg;
                long t1 = System.currentTimeMillis();
                boolean result = defaultMessageStore.putMessage(offset, message.getByteBodys());
                System.out.println("result:" + result);
                if (result) {
                    offset = getCommitOffset();
                    long t2 = System.currentTimeMillis();
                    System.out.println("消息保存成功，时间：" + ((t2 - t1) / 1000));
                    //上报位移
                    haClient.reportOffset();
                    System.out.println("上报位移成功");
                }
            }
        }

        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            System.out.println("slave节点与服务端断开连接了！");
            cause.printStackTrace();
            ctx.close();
            slaveOk = false;
        }

    }
}
